//
// Created by martin on 2/8/22.
//
#include "EventLoop.h"
#include "muduo/net/Poller.h"
#include "muduo/net/TimerQueue.h"
#include "muduo/net/TimerId.h"
#include "muduo/net/Channel.h"
#include "muduo/net/SocketsOps.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Timestamp.h"

#include <signal.h>
#include <functional>
#include <sys/eventfd.h>

using namespace muduo;
using namespace muduo::net;

namespace
{
__thread EventLoop* t_loopInThisThread = 0; // 指向当前线程创建的EventLoop对象

const int kPollTimeMs = 10000;

int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_SYSERR << "Failed in eventfd";
        abort();
    }
    return evtfd;
}

#if defined(__GNUC__)
#pragma GCC diagnostic ignored "-Wold-style-cast"
#endif
class IgnoreSigPipe
{
public:
    IgnoreSigPipe()
    {
        ::signal(SIGPIPE, SIG_IGN);
        // LOG_TRACE << "Ignore SIGPIPE";
    }
};
#if defined(__GNUC__)
#pragma GCC diagnostic error "-Wold-style-cast"
#endif

IgnoreSigPipe initObj;
} // namespace

EventLoop *EventLoop::getEventLoopOfCurrentThread()
{
    return t_loopInThisThread;
}

EventLoop::EventLoop()
: looping_(false),
quit_(false),
eventHandling_(false),
callingPendingFunctors_(false),
iteration_(0),
threadId_(CurrentThread::tid()),
poller_(Poller::newDefaultPoller(this)),
timerQueue_(new TimerQueue(this)),
wakeupFd_(createEventfd()),
wakeupChannel_(new Channel(this, wakeupFd_)),
currentActiveChannel_(NULL)
{
    LOG_DEBUG << "EventLoop create " << this << " in thread " << threadId_;
    if (t_loopInThisThread)
    {
        LOG_FATAL << "Another EventLoop " << t_loopInThisThread
        << " exists in this thread " << threadId_;
    }
    else
    {
        t_loopInThisThread = this;
    }
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // we are always reading the wakeupfd
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
    << " destructs in thread " << CurrentThread::tid();
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = NULL;
}

/**
 *  真正的工作循环.
 *  获得所有当前激活事件的通道，用Poller->poll()填到activeChannels_,
 *  然后调用Channel::handleEvent()处理每个激活通道.
 *
 *  最后排队运行所有pending函数, 通常是其他线程通过loop来调用运行用户任务
 */
void EventLoop::loop()
{
    assert(!looping_);    // to avoid reduplicate loop
    assertInLoopThread(); // to avoid new EventLoop() and loop() are not one thread
    looping_ = true;
    quit_ = false; // FIXME: what if someone calls quit() before loop() ?
    LOG_TRACE << "EventLoop " << this << " start looping";

    while (!quit_)
    {
        activeChannels_.clear();
        // 监听所有通道, 可能阻塞线程, 所有激活事件对应通道会填入activeChannels_
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
        ++iteration_; // 循环次数+1
        if (Logger::logLevel() <= Logger::TRACE)
        {
            printActiveChannels();
        }
        // TODO sort channel by priority
        // 处理所有激活事件

        eventHandling_ = true;
        for (Channel* channel : activeChannels_)
        {
            currentActiveChannel_ = channel;
            // 通过Channel::handleEvent回调事件处理函数
            currentActiveChannel_->handleEvent(pollReturnTime_);
        }
        currentActiveChannel_ = NULL;
        eventHandling_ = false;

        // 运行pending函数, 由其他线程请求调用的用户任务
        doPendingFunctors();
    }

    LOG_TRACE << "EventLoop " << this << " stop looping";
    looping_ = false;
}

/**
 * 推出loop循环
 */
void EventLoop::quit()
{
    quit_ = true;
    /*
     * There is a chance that loop() just executes while(!quit_) and exits,
     * then EventLoop destructs, then we are accessing an invalid object.
     * Can be fixed using mutex_ in both places.
     */
    if (!isInLoopThread())
    {
        wakeup(); // 线程可能阻塞在loop()中的Poller::poll()
    }
}

/**
 * 执行用户任务
 * @param cb 用户任务函数
 * @note 可以被多个线程执行。
 * 如果当前线程是创建当前EventLoop对象的线程，直接执行；
 * 否则，用户任务函数入队列pendingFunctors_成为一个pending functor，在loop循环中排队执行
 */
void EventLoop::runInLoop(Functor cb)
{
    if (isInLoopThread())
    {
        cb();
    }
    else
    {
        queueInLoop(std::move(cb));
    }
}

/**
 * 排队进入pendingFunctors_，等待执行
 * @param cb 用户任务函数
 * @note 可以被多个线程执行。
 * 如果当前线程不是创建当前EventLoop对象的线程 或者正在调用pending functor，
 * 就唤醒loop线程，避免loop线程阻塞.
 */
void EventLoop::queueInLoop(Functor cb)
{
    {
        MutexLockGuard lock(mutex_);
        pendingFunctors_.push_back(std::move(cb));
    }

    if (!isInLoopThread() || callingPendingFunctors_)
    { // callingPendingFunctors_为true表示正在处理pending functors, pendingFunctors_已经swap到函数局部变量中
        wakeup();
    }
}

size_t EventLoop::queueSize() const
{
    MutexLockGuard lock(mutex_);
    return pendingFunctors_.size();
}

/**
 * 定时功能，由用户指定绝对时间
 * @details 每为定时器队列timerQueue添加一个Timer,
 * timerQueue内部就会新建一个Timer对象, TimerId就保含了这个对象的唯一标识(序列号)
 * @param time 时间戳对象, 单位1us
 * @param cb 超时回调函数. 当前时间超过time代表时间时, EventLoop就会调用cb
 * @return 一个绑定timerQueue内部新增的Timer对象的TimerId对象, 用来唯一标识该Timer对象
 */
TimerId EventLoop::runAt(Timestamp time, TimerCallback cb)
{
    return timerQueue_->addTimer(std::move(cb), time, 0.0);
}

/**
 * 定时功能, 由用户相对时间, 通过runAt实现
 * @param delay 相对时间, 单位s, 精度1us(小数)
 * @param cb 超时回调
 */
TimerId EventLoop::runAfter(double delay, TimerCallback cb)
{
    Timestamp time(addTime(Timestamp::now(), delay));
    return runAt(time, std::move(cb));
}

/**
 * 定时功能, 由用户指定周期, 重复运行
 * @param interval 运行周期, 单位s, 精度1us(小数)
 * @param cb 超时回调
 * @return 一个绑定timerQueue内部新增的Timer对象的TimerId对象, 用来唯一标识该Timer对象
 */
TimerId EventLoop::runEvery(double interval, TimerCallback cb)
{
    Timestamp time(addTime(Timestamp::now(), interval));
    return timerQueue_->addTimer(std::move(cb), time, interval);
}

/**
 * 取消指定定时器
 * @param timerId Timer id, 唯一对应一个Timer对象
 */
void EventLoop::cancel(TimerId timerId)
{
    return timerQueue_->cancel(timerId);
}

/**
 * 根据具体poller对象, 来更新通道.
 * 会修改poller对象监听的通道数组.
 * @note 必须在channel所属loop线程运行
 */
void EventLoop::updateChannel(Channel *channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    poller_->updateChannel(channel);
}

/**
 * 根据具体poller对象, 来删除通道.
 * 会删除poller对象监听的通道数组.
 * @note 如果待移除通道正在激活事件队列, 应该先从激活事件队列中移除
 */
void EventLoop::removeChannel(Channel *channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    if (eventHandling_)
    {
        assert(currentActiveChannel_ == channel ||
        std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
    }
    poller_->removeChannel(channel);
}

/**
 * 判断poller是否正在监听通道channel
 * @note 必须在channel所属loop线程运行
 */
bool EventLoop::hasChannel(Channel *channel)
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    return poller_->hasChannel(channel);
}

void EventLoop::abortNotInLoopThread()
{
    LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
    << " was created in threadId_ = " << threadId_
    << ", current thread id = " << CurrentThread::tid();
}

/**
 * 其他线程唤醒等待在wakeupFd_上的线程, 产生读就绪事件.
 * @note write将添加8byte数据到内部计数器. 被唤醒线程必须调用read读取8byte数据.
 */
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = sockets::write(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
    }
}

/**
 * 处理wakeupChannel_读事件
 * @note read wakeupfd_
 */
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = sockets::read(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
    }
}

void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        MutexLockGuard lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for (const Functor& functor : functors)
    {
        functor();
    }
    callingPendingFunctors_ = false;
}

void EventLoop::printActiveChannels() const
{
    for (const Channel* channel : activeChannels_)
    {
        LOG_TRACE << "{" << channel->reventsToString() << "} ";
    }
}
